package com.shujia.spark.core

import com.shujia.spark.util.HDFSUtil
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo18Action {
  def main(args: Array[String]): Unit = {

    val conf = new SparkConf()

    conf.setAppName("group")

    conf.setMaster("local")

    val sc = new SparkContext(conf)

    val studentRDD: RDD[String] = sc.textFile("data/students.txt")

    /**
      * action算子，每一个action算子都会产生一个job
      *
      */

    /**
      * count: 统计rdd的数据行数
      */
    val count: Long = studentRDD.count()

    println(count)

    /**
      * sum: 对rdd中的数据求和，rdd中的数据类型必须是数字
      *
      */

    val sumAge: Double = studentRDD
      .map(line => line.split(",")(2).toInt)
      //对所有数据求和，只能是数字类型
      .sum()

    println(sumAge / count)


    /**
      * take: 取top, 返回一个数组， 不能取太多， 会导致内存溢出
      *
      */

    val top: Array[String] = studentRDD.take(10)

    top.foreach(println)

    /**
      * collect: 将rdd转换成数组，如果rdd数据量比较大，会导致内存溢出  （1G）
      */


    val array: Array[String] = studentRDD.collect()

    array.foreach(println)


    /**
      * foreach: 遍历rdd中的数据，也是一个action算子
      * foreachPartition: 一次将一个分区的数据传递给后面的函数
      *
      */

    studentRDD.foreach(println)


    studentRDD.foreachPartition((iter: Iterator[String]) => {
      iter.foreach(println)
    })

    /**
      * saveAsTextFile: 将数据保存在hdfs中
      *
      */

    HDFSUtil.deletePath("data/test")
    studentRDD.saveAsTextFile("data/test")


    while (true) {}
  }
}
